package demo.main;

import demo.kafka.streams.utils.Constants;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

import java.util.Properties;

public class UserClickCounter {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "user-click-counter-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_BOOTSTRAP_SERVERS);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();

        // 1. 从输入Topic创建一个KStream
        KStream<String, String> clicks = builder.stream("user-clicks");

        // 2. 按key（用户名）分组
        KGroupedStream<String, String> groupedByUser = clicks.groupByKey();

        // 3. 统计每个key的数量（这是一个有状态的聚合操作，结果是一个KTable）
        KTable<String, Long> userClickCounts = groupedByUser.count(Materialized.as("user-click-counts-store"));

        // 4. 将结果输出到另一个Topic
        userClickCounts.toStream().to("user-click-counts", Produced.with(Serdes.String(), Serdes.Long()));

        Topology topology = builder.build();
        KafkaStreams streams = new KafkaStreams(topology, props);
        streams.start();

        // 添加关闭钩子，优雅关闭
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}
